package com.sita.redsun.connector.handler;

import com.google.common.io.BaseEncoding;
import com.sita.redsun.connector.session.SessionMap;
import com.sita.redsun.connector.session.SunSession;
import com.sita.redsun.other.MQActive;
import com.sita.redsun.mq.MQSender;
import com.sita.redsun.util.AppContexConfig;
import io.netty.channel.*;
import io.netty.handler.codec.mqtt.*;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.AttributeKey;
import org.springframework.beans.factory.annotation.Autowired;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
 * Created by zhebinwang on 2017/3/28.
 *
 * 这个类负责接收SOCKET数据， 并且每个数据都已经是一个完整并单独的协议包了
 * SimpleChannelInboundHandler
 */
public class  EchoServerHandler extends SimpleChannelInboundHandler<MqttMessage> {
//public class EchoServerHandler  extends ChannelInboundHandlerAdapter {
    private long reciveNumber=0;
    private int handlerNumber=0;

    //因为在channel active 的时候我们还不知道session的身份，所以要定义一个成员
    //以方便我们随时更新他的身份。
    private SunSession sunSession;

    //标识这个session 自从建立后是否入map 如果没有入的话
    //那么在第一次的channelread 的时候入map
    private boolean isInMap=false;

    //貌似每个 ctx都有个 属性集合对象
    private AttributeKey<SunSession> sk = AttributeKey.valueOf("session");

    private static MQSender mqSender;
    private ExecutorService pool;
    @Autowired
    private MQActive mqActive;

    private static final Logger logger = LoggerFactory.getLogger(EchoServerHandler.class);

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

        int a=0;
        a++;
        String resultStr = BaseEncoding.base16().encode((byte[]) msg);
        logger.debug(resultStr);
        String gpsid=resultStr.substring(9,25);
        //每次连接 都记录这个session的最后传送数据时间
        if (sunSession!=null){
            sunSession.setLastActiveTimeToNow();
        }
        //首先判断session是否加入到map中去了，没加入就加入一下
        if (!isInMap && sunSession!=null){
            sunSession.setSessionUniqueId(gpsid);
            SessionMap.sharedInstance().addSession(gpsid,sunSession);
            logger.info(sunSession.getSessionUniqueId()+"已经被写入sunsession");
            isInMap=true;
        }

        //ChannelInboundHandlerAdapter
        //ctx.write(msg);
        //byte[] result = (byte[]) msg;
        //byte[] result1 = new byte[result.readableBytes()];
        //result.readBytes(result1);

        reciveNumber++;
        if (handlerNumber==0){
            handlerNumber=HandlerNumber.echoNumber++;
        }
        System.out.println("Client said:handlerNumber is "+handlerNumber +" reciveNumber"+reciveNumber+" info is:" + resultStr);

        //MQSender.sendMessage(resultStr);
        handlerSend(resultStr);

        //此处需要返回回应包
        String tmpstr=resultStr.substring(6,8);
        if (tmpstr!=null && tmpstr.equals("19")){
            //19号协议回应
            String returnstr=resultStr.substring(0,24)+"000D0A";
            //byte[] rebytes=BaseEncoding.base16().decode(returnstr);

            ctx.writeAndFlush(returnstr);
        }





        logger.info( "11");
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, MqttMessage mqttMessage) throws Exception {

    }

/*
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object o) throws Exception {

    }*/

    //真实的业务处理在这里
    private void handlerSend(final String resultStr){
        if (mqSender == null) {
            mqSender = (MQSender) AppContexConfig.getBean("mqsender");
            //mqSender=new MQSender();
        }
        if (pool == null) {
            int tpCore = Integer.valueOf(System.getProperty("queue.sender.threadpool.core.size", "8"));
            int tpMax = Integer.valueOf(System.getProperty("queue.sender.threadpool.max.size", "256"));
            //logger.debug("CONNECTOR_THREAD_POOL core-size={}, max-size={}", tpCore, tpMax);
            pool = new ThreadPoolExecutor(
                    tpCore, tpMax,
                    0L, TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue<Runnable>());
        }
        //TextMessage textMessage=new TextMessage();

        pool.submit(new Runnable() {
            @Override
            public void run() {
                //logger.debug("Begin sending MoMessage={} time={}", moMessage.hashCode(), System.currentTimeMillis());


                mqSender.send(resultStr);
            }
        });

    }



    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        //ctx.flush();
        logger.info( "22");
    }


    //可以理解为连接建立的时候调用的方法，我们在此处建立session 并将它
    //保存到sessionmap 中去
    @Override
    public void channelActive(final ChannelHandlerContext ctx) {
        isInMap=false;
        //logger.debug("CHANNEL_ACTIVE " + ctx.channel().remoteAddress());
        sunSession= new SunSession(ctx);
        sunSession.setLastActiveTimeToNow();
        //ctx.channel().attr(sk).set(sunSession);
        //SessionMap.sharedInstance().addSession(,sunSession);


      /*  MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.CONNACK,false,MqttQoS.AT_LEAST_ONCE,false,1);
        MqttConnAckVariableHeader variableHeader = new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED,false);
        MqttConnAckMessage conAck = new MqttConnAckMessage(mqttFixedHeader,variableHeader);
        ctx.channel().write(conAck);
        ctx.channel().flush();
        System.out.println(conAck);*/
    }




    //在框架中 处理任何异常出现的时候都将在此处进行异常处理
    //我们的任务就是把这个session删除掉
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        String key = null;
        try {
            if (sunSession != null) {
                key = sunSession.getSessionUniqueId();
                if (sunSession.isValid()) {
                    SessionMap.sharedInstance().removeSession(key);
                }
            }
        } catch (Exception e){

            e.printStackTrace();
        }
        if (cause instanceof IOException) {
            IOException ioe = (IOException) cause;
            if (ioe.getCause() == ioe) {
                if (ioe.getMessage().startsWith("Connection reset by peer")) {
                    logger.warn("CLIENT_CLOSE_CONNECTION for {}", key);
                } else
                    logger.error("CHANNEL_EXCEPTION " + cause);
            } else
                logger.error("CHANNEL_EXCEPTION " + cause);
        } else
            logger.error("CHANNEL_EXCEPTION " + cause);
        close(ctx);
    }

    /**
     * 处理服务端超市， 其中的 IdleStateEvent 标识超时事件
     * 共有三种类型， 写超时  读超时  读或者写超时，
       设置超时时间 是在初始化的时候通过
        ch.pipeline().addLast(new IdleStateHandler(30,30,7));
     这句话来进行的，其实就是增加了一个handler ，

     * 反正超时了就从sessionmap中移除 session  并且 关闭这个链接。
     *
     * **/
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent idle = (IdleStateEvent) evt;
            if (idle.state() == IdleState.ALL_IDLE) {
                // ctx.pipeline().get("idleStateHandler").
                logger.debug("CLIENT_IDLE {}", ctx.channel().remoteAddress());
                if (sunSession.isValid()) {
                    SessionMap.sharedInstance().removeSession(sunSession.getSessionUniqueId());
                }
                close(ctx);
            }

        }
    }
    /**
     * 断开连接的时候会激发 channelInactive
     * 事件 ，在这个调用里面我们把session删除
     * */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        //logger.debug("CHANNEL_INACTIVE " + ctx.channel().remoteAddress());
        if (sunSession.isValid()) {
            SessionMap.sharedInstance().removeSession(sunSession.getSessionUniqueId());
        }
        close(ctx);
    }


    private synchronized void close(ChannelHandlerContext ctx) {
        try {
            ChannelFuture future = ctx.close();
            future.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (future.isSuccess()) {
                        //logger.debug("Close connection: CHANNEL_CLOSED");
                    }
                }
            });
        } catch (Exception e) {
            logger.error("这是错误信息", e);
        }
    }

}
